package util;

import org.elasticsearch.client.Client;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.security.InvalidParameterException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * \* User: MeiZhongHao
 * \* Date: 2018-09-12
 * \* Time: 22:56
 * \* Description:
 * \
 */
public class ThreadReadFileHelper {

    public static List<Client> clients = new ArrayList<>();

    public static BlockingQueue<String> queue = new LinkedBlockingQueue<String>(12000);

    private volatile static boolean isDone = false;

    public static void main(String[] args) throws Exception {
        try {
            ThreadReadFileHelper helper = new ThreadReadFileHelper();
            helper.read("C:\\Users\\Administrator\\Desktop\\rt_risk_list.sql", Runtime.getRuntime().availableProcessors(), '\n');
            helper.writeData(5, clients);
            while (true) {
                if (queue.isEmpty()) {
                    //通知所有的FileWriteTask停止任务处理
                    helper.isDone = true;
                    System.out.println("========" + isDone + "========");
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            for (Client client : clients) {
                client.close();
            }
        }
    }

    public Client clientFactory(List<Client> clients) throws Exception {
//        Client client = new PreBuiltTransportClient(Settings.EMPTY)
//                .addTransportAddress(new TransportAddress(InetAddress.getByName("localhost"), 9300) {
//                });
//        clients.add(client);
        return null;
    }

    public void read(String path, int threadCount, char separator) throws IOException, InterruptedException {
        if (threadCount < 1) {
            throw new InvalidParameterException("The threadCount can not be less than 1");
        }
        if (path == null || path.isEmpty()) {
            throw new InvalidParameterException("The path can not be null or empty");
        }
        RandomAccessFile randomAccessFile = new RandomAccessFile(path, "r");
        long fileTotalLength = randomAccessFile.length();
        long gap = fileTotalLength / threadCount;
        long checkIndex = 0;
        long[] beginIndexs = new long[threadCount];
        long[] endIndexs = new long[threadCount];

        for (int n = 0; n < threadCount; n++) {
            beginIndexs[n] = checkIndex;
            if (n + 1 == threadCount) {
                endIndexs[n] = fileTotalLength;
                break;
            }
            checkIndex += gap;
            long gapToEof = getGapToEof(checkIndex, randomAccessFile, separator);
            checkIndex += gapToEof;
            endIndexs[n] = checkIndex;
        }

        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

        executorService.execute(() -> {
            try {
                readData(beginIndexs[0], endIndexs[0], path, randomAccessFile, CallbackFactory.getQueueCallback(queue));
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
        });

        for (int n = 1; n < threadCount; n++) {
            long begin = beginIndexs[n];
            long end = endIndexs[n];
            executorService.execute(() -> {
                try {
                    readData(begin, end, path, null, CallbackFactory.getQueueCallback(queue));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        System.out.println("======================= read end =======================");
    }

    private long getGapToEof(long beginIndex, RandomAccessFile randomAccessFile, char separator) throws IOException {
        randomAccessFile.seek(beginIndex);
        long count = 0;
        while (randomAccessFile.read() != separator) {
            count++;
        }
        count++;
        return count;
    }

    private void readData(long begin, long end, String path, RandomAccessFile randomAccessFile, ReadCallBack callback) throws FileNotFoundException, IOException {
        System.out.println("开始工作" + Thread.currentThread().getName());
        if (randomAccessFile == null) {
            randomAccessFile = new RandomAccessFile(path, "r");
        }
        randomAccessFile.seek(begin);
        byte[] bytes = null;
        while (true) {
            try {
                bytes = randomAccessFile.readLine().getBytes("ISO-8859-1");
                callback.callback(bytes);
            } catch (Exception e) {
                System.out.println("error");
                break;
            }
            begin += bytes.length;
            if (begin >= end) {
                break;
            }
        }
        randomAccessFile.close();
    }

    public void writeData(int threadCount, List<Client> clients) throws Exception {
        if (threadCount <= 0) {
            return;
        }
        final CountDownLatch endGate = new CountDownLatch(threadCount);
        ExecutorService service = Executors.newFixedThreadPool(threadCount);
        AtomicInteger integer = new AtomicInteger();
        for (int i = 0; i < threadCount; i++) {
            Client client = clientFactory(clients);
            service.execute(() -> {
                String data = null;
                System.out.println(isDone);
                //这里会多做几次尝试，直到master置位isDone=true之后，再行退出
                while (!isDone) {
                    try {
                        data = queue.poll(50, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if (data == null) {
                        continue;
                    }
                    try {
                        data = data.replaceAll("\'", "");
                        String[] strings = data.split(",");
                        Map<String, Object> json = new HashMap<String, Object>();
                        json.put("mobile", strings[2]);
                        json.put("id_card", strings[3]);
                        json.put("name", strings[4]);
                        json.put("related_person", strings[5]);
                        json.put("risk_type", strings[6]);
                        json.put("variable_name", strings[7]);
                        json.put("variable_value", strings[8]);
                        json.put("risk_date", strings[9]);
                        json.put("load_time", strings[10]);
                        System.out.println(integer.getAndIncrement() + " : " + client.prepareIndex("loan", "investor", strings[0]).setSource(json).get().status().getStatus());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                endGate.countDown();
                System.out.println(Thread.currentThread().getName() + "---> countDown");
            });
        }
        endGate.await();
        System.out.println("==================== write end ====================");
    }

    public static abstract class ReadCallBack {
        private String charsetName = "utf-8";
        public abstract void callback(byte[] bytes);
    }

    public static class QueueCallback extends ReadCallBack {
        private BlockingQueue<Object> queue;

        public QueueCallback(String charsetName, BlockingQueue queue) {
            super.charsetName = charsetName;
            this.queue = queue;
        }

        public QueueCallback(BlockingQueue queue) {
            this.queue = queue;
        }

        public void callback(byte[] bytes) {
            try {
                queue.put(new String(bytes, super.charsetName));
                System.out.println(" queue put success");
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }

    public static class CallbackFactory {

        public static QueueCallback getQueueCallback(String charsetName, BlockingQueue queue) {
            return new QueueCallback(charsetName, queue);
        }

        public static QueueCallback getQueueCallback(BlockingQueue queue) {
            return new QueueCallback(queue);
        }
    }
}